<?php

namespace dengje\workerman\process\queue;

use think\Exception;
use think\facade\Cache;
use Workerman\Timer;
use Workerman\Worker;
use think\facade\Db;

class Client2
{
    protected $redis;
    private string $prefix = '';

    private $payload;


    /**
     * 默认订阅选项
     * @var array|int[]
     */
    private array $defaultOption = [
        'max_retry' => 3, // 最大重试次数
        'retry_interval' => 5, // 重试间隔s
        'job' => 'job' // 执行方法名
    ];

    protected $deleted = false;

    public function __construct()
    {
        // $this->worker = $worker;
        $this->redis = Cache::store('redis')->handler();
        $this->prefix = config('cache.stores.redis.prefix') ?? 'xinxuan';
    
       
    }

    /**
     * 获取完整key
     * @param $keyVarName
     * @return string
     */
    private function getKey($keyVarName, $queue = 'default'): string
    {
        $keys = [
            // 存放所有监听的队列完整名称
            'queue' => "{{$this->prefix}queue:{$queue}}",
            // 延迟队列完整key
            'delay' => "{{$this->prefix}queue:{$queue}:delay}",
            // 失败存放队列完整key
            'fail' => "{{$this->prefix}queue:{$queue}:fail}",
        ];

        return $keys[$keyVarName];
    }

    /**
     * 
     * 发布队列
     * @param mixed $job
     * @param mixed $data
     * @param mixed $queue
     * @return void
     */
    public function push($job, $data = null, $delay = 0, $queue = null)
    {
        // {"job":"\\abc\\a","maxTries":null,"timeout":null,"data":"","id":"dJMECBDIgXYrjuDzcuDrfSaSciNHEEZB","attempts":0}
        $time = time();
        $data = $this->jsonParse([
            'job' => $job,
            'data' => $data,
            'max_tries' => $this->defaultOption['max_retry'],
            'timeout' => null,
            'id' => uniqid(),
            'attempts' => 0
        ]);

        // 判断是否延迟队列
        if ($delay > 0) {
            $this->redis->zAdd($this->getKey('delay'), $time + $delay, $data);
        } else {
            $this->redis->RPUSH($this->getKey('queue'), $data);
        }

    }


    /**
     * 监听队列
     * @return void
     */
    public function run()
    {
        // 0.001
        Timer::add(0.000001, function () {

        
                
            
            // 移出列表的队列第一个元素
            $data = $this->redis->BLPOP($this->getKey('queue'), 1);

            if ($data) {
                $key = $data[0];
                $dataStr = $data[1];
                $jobs = $this->jsonParse($dataStr);
                // 执行方法
                try {
                    $this->payload = $jobs;
                    $this->ping();
                    app($jobs['job'])->fire($this, $jobs['data']);
                    // throw new Exception('1');
                } catch (\Throwable | \Exception | \Error $e) {
                    // 重试
                    if ($jobs['attempts'] < $jobs['max_tries']) {
                        $jobs['attempts']++;
                        $dataStr = $this->jsonParse($jobs);
                        $this->redis->RPUSH($this->getKey('queue'), $dataStr);
                    } else {
                        // 失败
                        $this->fail($dataStr);
                    }
                }


            }

        
        });
        $this->delayQueue();
    }

    private function ping()
    {
        // 尝试执行一个简单的查询来检查连接是否活跃
        try {
            Db::execute('SELECT 1');
        } catch (\PDOException $e) {
            // 如果查询失败，关闭当前连接
            Db::close();
            // 尝试重新连接
            Db::connect();
        }
    }

    /**
     * 延迟队列
     */
    private function delayQueue()
    {

        Timer::add(1, function () {
            // var_dump('延迟队列');
            $scoreArr = $this->redis->zRevRangeByScore($this->getKey('delay'), time(), '-inf', ['LIMIT', 0, 128]);
            if (!empty($scoreArr)) {
                foreach ($scoreArr as $dataStr) {
                    $result = $this->redis->zRem($this->getKey('delay'), $dataStr);
                    if ($result !== 1) {
                        continue;
                    }



                    try {
                        $this->redis->lPush($this->getKey('queue'), $dataStr);

                    } catch (\Throwable | \Exception | \Error $e) {

                        // 重试
                    // if ($jobs['attempts'] < $jobs['max_tries']) {
                    //     $jobs['attempts']++;
                    //     $dataStr = $this->jsonParse($jobs);
                    //     $this->redis->RPUSH($this->getKey('queue'), $dataStr);
                    // } else {
                    //     // 失败
                    //     $this->fail($dataStr);
                    // }
                        continue;
                    }
                }
            }

        });
    }

    public function attempts()
    {
        return $this->payload['attempts'];
    }

    /**
     * 删除任务
     */
    public function delete()
    {
        $this->deleted = true;

    }

    /**
     * 保存失败任务
     * @param mixed $job
     * @return void
     */
    private function fail($job)
    {
        $this->redis->RPUSH($this->getKey('fail'), $job);
    }


    private function jsonParse($data)
    {
        return is_string($data) ? json_decode($data, JSON_UNESCAPED_UNICODE) : json_encode($data, true);
    }

}